//! # 文件存储实现
//!
//! 基于文件系统的消息持久化存储实现

use async_trait::async_trait;
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use std::path::PathBuf;
use std::sync::Arc;
use tokio::fs;
use tokio::sync::{Mutex, RwLock};
use uuid::Uuid;

use crate::error::{MessageBusError, MessageBusResult};
use crate::message::Message;
use crate::persistence::{
    MessageRecord, MessageStatus, MessageStorage, OrderBy, QueryCriteria, 
    SortOrder, StorageStats, StorageTransaction, TimeRange
};

/// 文件存储配置
#[derive(Debug, Clone)]
pub struct FileStorageConfig {
    /// 存储目录
    pub storage_dir: PathBuf,
    /// 是否启用压缩
    pub enable_compression: bool,
    /// 文件分片大小（字节）
    pub shard_size: usize,
    /// 索引缓存大小
    pub index_cache_size: usize,
    /// 自动清理间隔（秒）
    pub cleanup_interval_secs: u64,
}

/// 文件存储实现
pub struct FileStorage {
    config: FileStorageConfig,
    index: Arc<RwLock<MessageIndex>>,
    transactions: Arc<Mutex<HashMap<String, FileTransaction>>>,
    stats: Arc<RwLock<StorageStats>>,
}

/// 消息索引
#[derive(Debug, Default)]
struct MessageIndex {
    /// 消息ID到文件位置的映射
    id_to_location: HashMap<String, FileLocation>,
    /// 状态索引
    status_index: HashMap<MessageStatus, Vec<String>>,
    /// 主题索引
    topic_index: HashMap<String, Vec<String>>,
}

/// 文件位置信息
#[derive(Debug, Clone)]
struct FileLocation {
    /// 文件路径
    file_path: PathBuf,
    /// 文件内偏移量
    offset: u64,
    /// 数据长度
    length: u64,
}

/// 文件事务
struct FileTransaction {
    id: String,
    operations: Vec<TransactionOp>,
    created_at: chrono::DateTime<chrono::Utc>,
}

/// 事务操作
#[derive(Debug, Clone)]
enum TransactionOp {
    Store(MessageRecord),
    UpdateStatus(String, MessageStatus),
    Delete(String),
}

impl FileStorage {
    /// 创建新的文件存储
    pub async fn new(config: FileStorageConfig) -> MessageBusResult<Self> {
        tracing::info!(
            "正在创建文件存储: 目录={}, 分片大小={} MB, 缓存大小={}",
            config.storage_dir.display(),
            config.shard_size / (1024 * 1024),
            config.index_cache_size
        );
        
        // 创建存储目录
        if let Err(e) = fs::create_dir_all(&config.storage_dir).await {
            let error_msg = format!("创建存储目录失败: {}", e);
            tracing::error!("{}", error_msg);
            return Err(MessageBusError::storage_error(error_msg));
        }
        tracing::debug!("存储目录创建成功: {}", config.storage_dir.display());

        let storage = Self {
            config: config.clone(),
            index: Arc::new(RwLock::new(MessageIndex::default())),
            transactions: Arc::new(Mutex::new(HashMap::new())),
            stats: Arc::new(RwLock::new(StorageStats::default())),
        };

        // 加载现有数据
        tracing::debug!("开始加载现有数据...");
        storage.load_existing_data().await?;

        // 启动清理任务
        tracing::debug!("启动文件存储清理任务");
        storage.start_cleanup_task();
        
        tracing::info!("文件存储创建完成");

        Ok(storage)
    }

    /// 加载现有数据
    async fn load_existing_data(&self) -> MessageBusResult<()> {
        // 实现数据加载逻辑
        tracing::info!("加载现有消息数据...");
        Ok(())
    }

    /// 启动清理任务
    fn start_cleanup_task(&self) {
        let storage_weak = Arc::downgrade(&Arc::new(self.clone()));
        let interval_secs = self.config.cleanup_interval_secs;

        tokio::spawn(async move {
            let mut interval = tokio::time::interval(
                std::time::Duration::from_secs(interval_secs)
            );

            loop {
                interval.tick().await;
                
                if let Some(storage) = storage_weak.upgrade() {
                    if let Err(e) = storage.cleanup_expired().await {
                        tracing::error!("清理过期消息失败: {}", e);
                    }
                } else {
                    break;
                }
            }
        });
    }
}

#[async_trait]
impl MessageStorage for FileStorage {
    async fn store(&self, message: Message) -> MessageBusResult<String> {
        let start_time = std::time::Instant::now();
        let record = MessageRecord::new(message);
        let id = record.id.clone();
        
        tracing::debug!(
            "开始存储消息: id={}, topic={}, size={} bytes",
            id,
            record.message.topic(),
            record.message.size()
        );
        
        // 写入文件
        let location = match self.write_record(&record).await {
            Ok(loc) => {
                tracing::trace!(
                    "消息写入文件成功: id={}, file={}",
                    id,
                    loc.file_path.display()
                );
                loc
            }
            Err(e) => {
                tracing::error!("消息写入文件失败: id={}, error={}", id, e);
                return Err(e);
            }
        };
        
        // 更新索引
        {
            let mut index = self.index.write().await;
            if let Err(e) = index.add_record(&record) {
                tracing::error!("更新索引失败: id={}, error={}", id, e);
                return Err(e);
            }
            // 添加文件位置索引
            index.add_location(&id, location);
            tracing::trace!("索引更新成功: id={}", id);
        }
        
        // 更新统计
        {
            let mut stats = self.stats.write().await;
            stats.total_messages += 1;
            stats.pending_messages += 1;
            
            if stats.oldest_message.is_none() {
                stats.oldest_message = Some(record.stored_at);
            }
            stats.newest_message = Some(record.stored_at);
        }
        
        let duration = start_time.elapsed();
        tracing::info!(
            "消息存储成功: id={}, topic={}, duration={:?}",
            id,
            record.message.topic(),
            duration
        );
        
        Ok(id)
    }

    async fn store_batch(&self, messages: Vec<Message>) -> MessageBusResult<Vec<String>> {
        let mut ids = Vec::new();
        for message in messages {
            let id = self.store(message).await?;
            ids.push(id);
        }
        Ok(ids)
    }

    async fn get(&self, message_id: &str) -> MessageBusResult<Option<MessageRecord>> {
        let index = self.index.read().await;
        if let Some(location) = index.id_to_location.get(message_id) {
            self.read_record(location).await.map(Some)
        } else {
            Ok(None)
        }
    }

    async fn query(&self, criteria: QueryCriteria) -> MessageBusResult<Vec<MessageRecord>> {
        let start_time = std::time::Instant::now();
        
        tracing::debug!(
            "开始查询消息: status={:?}, topic_pattern={:?}, limit={:?}",
            criteria.status,
            criteria.topic_pattern,
            criteria.limit
        );
        
        let index = self.index.read().await;
        let mut matching_ids = Vec::new();

        // 根据状态过滤
        if let Some(status) = &criteria.status {
            if let Some(ids) = index.status_index.get(status) {
                matching_ids.extend(ids.clone());
                tracing::trace!("根据状态查找到 {} 条消息: status={:?}", ids.len(), status);
            }
        } else {
            matching_ids.extend(index.id_to_location.keys().cloned());
            tracing::trace!("查找所有消息: {} 条", matching_ids.len());
        }
        
        // 根据主题模式过滤
        if let Some(pattern) = &criteria.topic_pattern {
            let mut topic_filtered_ids = Vec::new();
            
            if pattern.contains('*') {
                // 通配符匹配
                for topic in index.topic_index.keys() {
                    if self.matches_topic_pattern(pattern, topic) {
                        if let Some(ids) = index.topic_index.get(topic) {
                            topic_filtered_ids.extend(ids.clone());
                        }
                    }
                }
            } else {
                // 精确匹配
                if let Some(ids) = index.topic_index.get(pattern) {
                    topic_filtered_ids.extend(ids.clone());
                }
            }
            
            // 取交集
            matching_ids.retain(|id| topic_filtered_ids.contains(id));
            tracing::trace!("根据主题过滤后剩余 {} 条消息", matching_ids.len());
        }

        // 加载消息记录
        let mut records = Vec::new();
        let mut loaded_count = 0;
        let mut error_count = 0;
        
        for id in matching_ids {
            if let Some(limit) = criteria.limit {
                if records.len() >= limit {
                    break;
                }
            }
            
            if let Some(location) = index.id_to_location.get(&id) {
                match self.read_record(location).await {
                    Ok(record) => {
                        // 时间范围过滤
                        if let Some(ref time_range) = criteria.time_range {
                            if record.stored_at < time_range.start || record.stored_at > time_range.end {
                                continue;
                            }
                        }
                        
                        records.push(record);
                        loaded_count += 1;
                    }
                    Err(e) => {
                        tracing::warn!("加载消息记录失败: id={}, error={}", id, e);
                        error_count += 1;
                    }
                }
            }
        }
        
        // 排序
        if let Some(ref order_by) = criteria.order_by {
            match order_by {
                OrderBy::StoredAt(sort_order) => {
                    match sort_order {
                        SortOrder::Ascending => records.sort_by(|a, b| a.stored_at.cmp(&b.stored_at)),
                        SortOrder::Descending => records.sort_by(|a, b| b.stored_at.cmp(&a.stored_at)),
                    }
                }
                OrderBy::MessageId(sort_order) => {
                    match sort_order {
                        SortOrder::Ascending => records.sort_by(|a, b| a.id.cmp(&b.id)),
                        SortOrder::Descending => records.sort_by(|a, b| b.id.cmp(&a.id)),
                    }
                }
                OrderBy::RetryCount(sort_order) => {
                    match sort_order {
                        SortOrder::Ascending => records.sort_by(|a, b| a.retry_count.cmp(&b.retry_count)),
                        SortOrder::Descending => records.sort_by(|a, b| b.retry_count.cmp(&a.retry_count)),
                    }
                }
            }
            tracing::trace!("消息排序完成: order_by={:?}", order_by);
        }
        
        // 应用偏移量
        if let Some(offset) = criteria.offset {
            if offset < records.len() {
                records.drain(0..offset);
            } else {
                records.clear();
            }
        }
        
        let duration = start_time.elapsed();
        tracing::info!(
            "查询完成: 返回={} 条, 加载成功={}, 加载失败={}, duration={:?}",
            records.len(), loaded_count, error_count, duration
        );

        Ok(records)
    }

    async fn update_status(&self, message_id: &str, status: MessageStatus) -> MessageBusResult<()> {
        // 实现状态更新逻辑
        let mut index = self.index.write().await;
        index.update_status(message_id, status)?;
        Ok(())
    }

    async fn delete(&self, message_id: &str) -> MessageBusResult<()> {
        let mut index = self.index.write().await;
        index.remove_record(message_id)?;
        Ok(())
    }

    async fn delete_batch(&self, message_ids: Vec<String>) -> MessageBusResult<()> {
        for id in message_ids {
            self.delete(&id).await?;
        }
        Ok(())
    }

    async fn cleanup_expired(&self) -> MessageBusResult<u64> {
        let start_time = std::time::Instant::now();
        tracing::debug!("开始清理过期消息");
        
        let now = chrono::Utc::now();
        let mut expired_ids = Vec::new();
        
        // 查找所有过期消息
        {
            let index = self.index.read().await;
            for id in index.id_to_location.keys() {
                if let Some(location) = index.id_to_location.get(id) {
                    match self.read_record(location).await {
                        Ok(record) => {
                            if record.is_expired() {
                                expired_ids.push(id.clone());
                                tracing::trace!("发现过期消息: id={}, expired_at={:?}", id, record.expires_at);
                            }
                        }
                        Err(e) => {
                            tracing::warn!("读取消息记录失败，将其标记为待清理: id={}, error={}", id, e);
                            expired_ids.push(id.clone());
                        }
                    }
                }
            }
        }
        
        // 清理过期消息
        let mut cleaned_count = 0u64;
        for id in &expired_ids {
            match self.delete(id).await {
                Ok(()) => {
                    cleaned_count += 1;
                    tracing::trace!("过期消息清理成功: id={}", id);
                }
                Err(e) => {
                    tracing::error!("清理过期消息失败: id={}, error={}", id, e);
                }
            }
        }
        
        let duration = start_time.elapsed();
        if cleaned_count > 0 {
            tracing::info!(
                "过期消息清理完成: 清理={} 条, 检查={} 条, duration={:?}",
                cleaned_count, expired_ids.len(), duration
            );
        } else {
            tracing::debug!(
                "未发现过期消息: 检查={} 条, duration={:?}",
                expired_ids.len(), duration
            );
        }
        
        Ok(cleaned_count)
    }

    async fn get_stats(&self) -> MessageBusResult<StorageStats> {
        let stats = self.stats.read().await;
        Ok(stats.clone())
    }

    async fn begin_transaction(&self) -> MessageBusResult<Box<dyn StorageTransaction>> {
        let tx_id = Uuid::new_v4().to_string();
        let tx = FileTransaction {
            id: tx_id.clone(),
            operations: Vec::new(),
            created_at: chrono::Utc::now(),
        };

        let mut transactions = self.transactions.lock().await;
        transactions.insert(tx_id.clone(), tx);

        Ok(Box::new(FileStorageTransaction {
            id: tx_id,
            storage: self.clone(),
        }))
    }

    async fn health_check(&self) -> MessageBusResult<()> {
        // 检查存储目录是否可访问
        if !self.config.storage_dir.exists() {
            return Err(MessageBusError::storage_error("存储目录不存在"));
        }
        Ok(())
    }
}

/// 文件存储事务实现
struct FileStorageTransaction {
    id: String,
    storage: FileStorage,
}

#[async_trait]
impl StorageTransaction for FileStorageTransaction {
    fn id(&self) -> &str {
        &self.id
    }

    async fn store(&mut self, message: Message) -> MessageBusResult<String> {
        let record = MessageRecord::new(message);
        let id = record.id.clone();
        
        // 添加到事务操作中
        let mut transactions = self.storage.transactions.lock().await;
        if let Some(tx) = transactions.get_mut(&self.id) {
            tx.operations.push(TransactionOp::Store(record));
        }
        
        Ok(id)
    }

    async fn update_status(&mut self, message_id: &str, status: MessageStatus) -> MessageBusResult<()> {
        let mut transactions = self.storage.transactions.lock().await;
        if let Some(tx) = transactions.get_mut(&self.id) {
            tx.operations.push(TransactionOp::UpdateStatus(message_id.to_string(), status));
        }
        Ok(())
    }

    async fn delete(&mut self, message_id: &str) -> MessageBusResult<()> {
        let mut transactions = self.storage.transactions.lock().await;
        if let Some(tx) = transactions.get_mut(&self.id) {
            tx.operations.push(TransactionOp::Delete(message_id.to_string()));
        }
        Ok(())
    }

    async fn commit(self: Box<Self>) -> MessageBusResult<()> {
        // 执行所有事务操作
        let mut transactions = self.storage.transactions.lock().await;
        if let Some(tx) = transactions.remove(&self.id) {
            for op in tx.operations {
                match op {
                    TransactionOp::Store(record) => {
                        // 写入文件
                        let location = self.storage.write_record(&record).await?;
                        
                        // 更新索引
                        {
                            let mut index = self.storage.index.write().await;
                            index.add_record(&record)?;
                            index.add_location(&record.id, location);
                        }
                        
                        // 更新统计
                        {
                            let mut stats = self.storage.stats.write().await;
                            stats.total_messages += 1;
                            stats.pending_messages += 1;
                            
                            if stats.oldest_message.is_none() {
                                stats.oldest_message = Some(record.stored_at);
                            }
                            stats.newest_message = Some(record.stored_at);
                        }
                    }
                    TransactionOp::UpdateStatus(id, status) => {
                        self.storage.update_status(&id, status).await?;
                    }
                    TransactionOp::Delete(id) => {
                        self.storage.delete(&id).await?;
                    }
                }
            }
        }
        Ok(())
    }

    async fn rollback(self: Box<Self>) -> MessageBusResult<()> {
        // 简单地移除事务
        let mut transactions = self.storage.transactions.lock().await;
        transactions.remove(&self.id);
        Ok(())
    }
}

impl FileStorage {
    /// 写入消息记录到文件
    async fn write_record(&self, record: &MessageRecord) -> MessageBusResult<FileLocation> {
        let start_time = std::time::Instant::now();
        
        // 获取分片文件路径
        let file_path = self.get_shard_path(&record.id);
        tracing::trace!("将消息写入分片文件: id={}, file={}", record.id, file_path.display());
        
        // 序列化消息记录
        let serialized = serde_json::to_vec(record)
            .map_err(|e| {
                tracing::error!("消息序列化失败: id={}, error={}", record.id, e);
                MessageBusError::serialization_error(e.to_string())
            })?;
        
        let data_size = serialized.len();
        tracing::trace!("消息序列化成功: id={}, size={} bytes", record.id, data_size);

        // 检查文件大小限制
        let current_size = if file_path.exists() {
            fs::metadata(&file_path).await
                .map_err(|e| MessageBusError::storage_error(format!("获取文件信息失败: {}", e)))?
                .len()
        } else {
            0
        };
        
        // 如果文件太大，创建新的分片
        let (actual_file_path, offset) = if current_size + data_size as u64 > self.config.shard_size as u64 {
            let new_file_path = self.get_new_shard_path(&record.id);
            tracing::debug!("分片文件超限，创建新分片: old={}, new={}", file_path.display(), new_file_path.display());
            (new_file_path, 0)
        } else {
            (file_path, current_size)
        };
        
        // 写入数据到文件
        if offset == 0 {
            // 新文件，直接写入
            fs::write(&actual_file_path, &serialized).await
        } else {
            // 追加到现有文件
            use tokio::io::AsyncWriteExt;
            let mut file = tokio::fs::OpenOptions::new()
                .create(true)
                .append(true)
                .open(&actual_file_path)
                .await
                .map_err(|e| MessageBusError::storage_error(format!("打开文件失败: {}", e)))?;
            
            file.write_all(&serialized).await
                .map_err(|e| MessageBusError::storage_error(format!("写入文件失败: {}", e)))?;
            
            file.sync_all().await
                .map_err(|e| MessageBusError::storage_error(format!("同步文件失败: {}", e)))?;
            
            Ok(())
        }.map_err(|e| {
            tracing::error!("文件写入失败: id={}, file={}, error={}", record.id, actual_file_path.display(), e);
            e
        })?;
        
        let location = FileLocation {
            file_path: actual_file_path.clone(),
            offset,
            length: data_size as u64,
        };
        
        let duration = start_time.elapsed();
        tracing::trace!("消息写入文件成功: id={}, file={}, offset={}, size={}, duration={:?}", 
            record.id, actual_file_path.display(), offset, data_size, duration);

        Ok(location)
    }

    /// 从文件读取消息记录
    async fn read_record(&self, location: &FileLocation) -> MessageBusResult<MessageRecord> {
        let start_time = std::time::Instant::now();
        
        tracing::trace!("开始读取消息: file={}, offset={}, length={}", 
            location.file_path.display(), location.offset, location.length);
        
        // 检查文件是否存在
        if !location.file_path.exists() {
            return Err(MessageBusError::storage_error(
                format!("消息文件不存在: {}", location.file_path.display())
            ));
        }
        
        let data = if location.offset == 0 && location.length == 0 {
            // 整个文件读取（兼容旧格式）
            fs::read(&location.file_path).await
                .map_err(|e| MessageBusError::storage_error(format!("读取文件失败: {}", e)))?
        } else {
            // 从指定位置读取指定长度
            use tokio::io::{AsyncReadExt, AsyncSeekExt};
            
            let mut file = tokio::fs::File::open(&location.file_path).await
                .map_err(|e| MessageBusError::storage_error(format!("打开文件失败: {}", e)))?;
            
            file.seek(std::io::SeekFrom::Start(location.offset)).await
                .map_err(|e| MessageBusError::storage_error(format!("文件定位失败: {}", e)))?;
            
            let mut buffer = vec![0u8; location.length as usize];
            file.read_exact(&mut buffer).await
                .map_err(|e| MessageBusError::storage_error(format!("读取数据失败: {}", e)))?;
            
            buffer
        };
        
        tracing::trace!("文件读取成功: size={} bytes", data.len());

        let record: MessageRecord = serde_json::from_slice(&data)
            .map_err(|e| {
                tracing::error!("消息反序列化失败: file={}, error={}", location.file_path.display(), e);
                MessageBusError::deserialization_error(e.to_string())
            })?;
        
        let duration = start_time.elapsed();
        tracing::trace!("消息读取完成: id={}, duration={:?}", record.id, duration);

        Ok(record)
    }

    /// 获取分片文件路径
    fn get_shard_path(&self, message_id: &str) -> PathBuf {
        let hash = self.hash_string(message_id);
        let shard_id = hash % 256; // 256个分片
        self.config.storage_dir.join(format!("shard_{:03}.dat", shard_id))
    }
    
    /// 获取新的分片文件路径（当当前分片太大时）
    fn get_new_shard_path(&self, message_id: &str) -> PathBuf {
        let hash = self.hash_string(message_id);
        let base_shard_id = hash % 256;
        let timestamp = chrono::Utc::now().timestamp();
        
        // 使用时间戳创建新的分片文件
        self.config.storage_dir.join(format!("shard_{:03}_{}.dat", base_shard_id, timestamp))
    }

    /// 简单哈希函数
    fn hash_string(&self, s: &str) -> usize {
        s.bytes().fold(0usize, |acc, b| acc.wrapping_mul(31).wrapping_add(b as usize))
    }
    
    /// 主题模式匹配
    fn matches_topic_pattern(&self, pattern: &str, topic: &str) -> bool {
        if pattern == "**" {
            return true; // 匹配所有
        }
        
        if !pattern.contains('*') {
            return pattern == topic; // 精确匹配
        }
        
        // 简单的通配符匹配
        let pattern_parts: Vec<&str> = pattern.split('*').collect();
        let mut topic_pos = 0;
        
        for (i, part) in pattern_parts.iter().enumerate() {
            if part.is_empty() {
                continue;
            }
            
            if let Some(pos) = topic[topic_pos..].find(part) {
                topic_pos += pos + part.len();
                
                // 如果是第一部分且不是以*开始，必须在开头匹配
                if i == 0 && !pattern.starts_with('*') && pos != 0 {
                    return false;
                }
                
                // 如果是最后一部分且不是以*结尾，必须在末尾匹配
                if i == pattern_parts.len() - 1 && !pattern.ends_with('*') {
                    return topic.ends_with(part);
                }
            } else {
                return false;
            }
        }
        
        true
    }
}

impl MessageIndex {
    /// 添加消息记录到索引
    fn add_record(&mut self, record: &MessageRecord) -> MessageBusResult<()> {
        tracing::trace!("添加消息到索引: id={}, topic={}, status={:?}", 
            record.id, record.message.topic(), record.status);
        
        // 添加到状态索引
        self.status_index
            .entry(record.status.clone())
            .or_insert_with(Vec::new)
            .push(record.id.clone());

        // 添加到主题索引
        self.topic_index
            .entry(record.message.topic().to_string())
            .or_insert_with(Vec::new)
            .push(record.id.clone());
        
        tracing::trace!("索引添加成功: id={}", record.id);
        Ok(())
    }
    
    /// 添加文件位置索引
    fn add_location(&mut self, message_id: &str, location: FileLocation) {
        tracing::trace!("添加文件位置索引: id={}, file={}", message_id, location.file_path.display());
        self.id_to_location.insert(message_id.to_string(), location);
    }

    /// 更新消息状态索引
    fn update_status(&mut self, message_id: &str, new_status: MessageStatus) -> MessageBusResult<()> {
        // 从旧状态索引中移除
        for (_, ids) in self.status_index.iter_mut() {
            ids.retain(|id| id != message_id);
        }

        // 添加到新状态索引
        self.status_index
            .entry(new_status)
            .or_insert_with(Vec::new)
            .push(message_id.to_string());

        Ok(())
    }

    /// 从索引中移除消息记录
    fn remove_record(&mut self, message_id: &str) -> MessageBusResult<()> {
        self.id_to_location.remove(message_id);
        
        for (_, ids) in self.status_index.iter_mut() {
            ids.retain(|id| id != message_id);
        }
        
        for (_, ids) in self.topic_index.iter_mut() {
            ids.retain(|id| id != message_id);
        }

        Ok(())
    }
}

impl Clone for FileStorage {
    fn clone(&self) -> Self {
        Self {
            config: self.config.clone(),
            index: self.index.clone(),
            transactions: self.transactions.clone(),
            stats: self.stats.clone(),
        }
    }
}

impl Default for FileStorageConfig {
    fn default() -> Self {
        Self {
            storage_dir: PathBuf::from("./message_storage"),
            enable_compression: true,
            shard_size: 64 * 1024 * 1024, // 64MB
            index_cache_size: 10000,
            cleanup_interval_secs: 3600, // 1小时
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use serde_json::json;
    use tempfile::TempDir;

    #[tokio::test]
    async fn test_file_storage_creation() {
        let temp_dir = TempDir::new().unwrap();
        let config = FileStorageConfig {
            storage_dir: temp_dir.path().to_path_buf(),
            ..Default::default()
        };

        let storage = FileStorage::new(config).await.unwrap();
        assert!(storage.health_check().await.is_ok());
    }

    #[tokio::test]
    async fn test_message_store_and_retrieve() {
        let temp_dir = TempDir::new().unwrap();
        let config = FileStorageConfig {
            storage_dir: temp_dir.path().to_path_buf(),
            ..Default::default()
        };

        let storage = FileStorage::new(config).await.unwrap();
        let message = Message::new("test.topic", json!({"test": "data"}));
        
        let message_id = storage.store(message.clone()).await.unwrap();
        let retrieved = storage.get(&message_id).await.unwrap();
        
        assert!(retrieved.is_some());
        let record = retrieved.unwrap();
        assert_eq!(record.message.topic(), "test.topic");
        assert_eq!(record.status, MessageStatus::Pending);
    }

    #[tokio::test]
    async fn test_transaction() {
        let temp_dir = TempDir::new().unwrap();
        let config = FileStorageConfig {
            storage_dir: temp_dir.path().to_path_buf(),
            ..Default::default()
        };

        let storage = FileStorage::new(config).await.unwrap();
        let mut tx = storage.begin_transaction().await.unwrap();
        
        let message = Message::new("test.topic", json!({"test": "data"}));
        let message_id = tx.store(message).await.unwrap();
        
        // 提交事务
        tx.commit().await.unwrap();
        
        // 提交后应该能查询到
        let stored_after_commit = storage.get(&message_id).await.unwrap();
        assert!(stored_after_commit.is_some());
        
        let record = stored_after_commit.unwrap();
        assert_eq!(record.message.topic(), "test.topic");
        assert_eq!(record.status, MessageStatus::Pending);
        
        // 测试事务回滚
        let mut tx2 = storage.begin_transaction().await.unwrap();
        let message2 = Message::new("test.rollback", json!({"rollback": true}));
        let message_id2 = tx2.store(message2).await.unwrap();
        
        // 回滚事务
        tx2.rollback().await.unwrap();
        
        // 回滚后消息应该不存在（或状态为回滚）
        let rolled_back = storage.get(&message_id2).await.unwrap();
        // 容忍不同的实现方式
        if let Some(record) = rolled_back {
            // 如果记录存在，状态应该反映回滚
            assert_ne!(record.status, MessageStatus::Completed);
        }
    }
}